package org.jetlinks.pro.things.impl.rpc;

import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.enums.ErrorCode;
import org.jetlinks.core.exception.DeviceOperationException;
import org.jetlinks.core.message.RepayableThingMessage;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.message.ThingMessageReply;
import org.jetlinks.core.things.ThingRpcSupport;
import org.jetlinks.pro.things.ThingConstants;
import org.jetlinks.pro.things.ThingMessageConverter;
import org.jetlinks.pro.things.utils.ThingsUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;

/**
 * 指定多个`物的RPC支持`,将这些RPC支持放到一起进行调用
 * 支持并行调用或串行调用
 *
 * @author zhouhao
 * @since 1.12
 */
@Slf4j
public class CompositeThingsRpcSupport implements ThingRpcSupport {

    /**
     * 多个RPC是否使用并行调用
     */
    @Setter
    private boolean parallel;

    /**
     * RPC过程中发生错误是否需要中断整个任务
     */
    @Setter
    private boolean interruptOnError = true;

    /**
     * 多个RPC动作集合
     */
    private final List<ThingRpcSupport> actions = new CopyOnWriteArrayList<>();

    /**
     * RPC结果转换器，用于将调用物RPC得到的数据转换为新的数据
     */
    @Setter
    private ThingMessageConverter responseConverter = ThingMessageConverter.identity();

    /**
     * RPC中的错误处理器
     */
    @Setter
    private Consumer<Throwable> errorHandler = err -> CompositeThingsRpcSupport.log.error(err.getMessage(), err);

    public CompositeThingsRpcSupport() {

    }

    /**
     * 添加RPC支持
     *
     * @param support RPC支持
     */
    public void addSupport(ThingRpcSupport support) {
        actions.add(support);
    }

    /**
     * 添加RPC支持
     *
     * @param index   RPC添加的序号，可用于支持RPC调用顺序
     * @param support RPC支持
     */
    public void addSupport(int index, ThingRpcSupport support) {
        actions.add(index, support);
    }

    /**
     * 消息调用结果包装，用于消息结果验证，消息错误处理
     *
     * @param result 物RPC结果流
     * @return 包装后的物消息流
     */
    private Flux<? extends ThingMessage> wrap(Flux<? extends ThingMessage> result) {
        if (interruptOnError) {
            return result
                .doOnNext(this::checkResponse);
        }
        return result
            .onErrorResume(err -> {
                errorHandler.accept(err);
                return Mono.empty();
            });
    }

    /**
     * 验证RPC结果，如果没有正常返回或调用异常抛出{@link DeviceOperationException}
     *
     * @param response RPC调用结果
     */
    private void checkResponse(ThingMessage response) {
        if (response instanceof ThingMessageReply) {
            ThingMessageReply reply = ((ThingMessageReply) response);
            if (!reply.isSuccess()) {
                String message = (String) reply.getHeaderOrElse("errorMessage", null);
                ErrorCode code = ErrorCode.of(reply.getCode())
                                          .orElse(ErrorCode.SYSTEM_ERROR);
                throw new DeviceOperationException(code, message == null ? code.getText() : message);
            }
        }
    }

    /**
     * RPC调用
     *
     * @param message RPC请求消息
     * @return 多个RPC的调用结果
     */
    @Override
    public Flux<? extends ThingMessage> call(ThingMessage message) {
        if (actions.size() == 0) {
            return Flux.empty();
        }
        ThingMessage request = message.copy();
        ThingMessage[] responses = new ThingMessage[actions.size()];
        //并行执行多个RPC
        if (parallel) {
            return Flux
                .fromIterable(actions)
                .index()
                .flatMap(tp2 -> tp2
                             .getT2()
                             .call(request)
                             .as(this::wrap)
                             //调用结果赋值
                             .doOnNext(res -> responses[tp2.getT1().intValue()] = res),
                         Integer.MAX_VALUE)
                //调用结果消息数据转换
                .thenMany(Flux.defer(() -> doConvert(request, responses)));
        }
        //串行执行多个RPC
        Flux<ThingMessage> caller = null;
        for (int i = 0, size = actions.size(); i < size; i++) {
            int fi = i;
            ThingRpcSupport support = actions.get(i);
            if (caller == null) {
                caller = support
                    .call(message)
                    .as(this::wrap)
                    .cast(ThingMessage.class)
                    .doOnNext(t -> responses[fi] = t);
            } else {
                caller = caller
                    .thenMany(
                        Flux.defer(() -> support
                            .call(createInvokeMessage(request, fi, responses))
                            .as(this::wrap)
                            .cast(ThingMessage.class)
                            .doOnNext(t -> responses[fi] = t)
                        )
                    );
            }
        }
        if (caller == null) {
            return Flux.empty();
        }
        return caller
            .thenMany(Flux.defer(() -> doConvert(request, responses)));
    }

    /**
     * 消息数据转换，例如RPC结果转换为功能回复的输出参数
     *
     * @param request   RPC请求消息
     * @param responses RPC结果集
     * @return 新的消息
     */
    private Flux<ThingMessage> doConvert(ThingMessage request, ThingMessage[] responses) {
        if (request instanceof RepayableThingMessage) {
            ThingMessageReply reply = ((RepayableThingMessage<?>) request).newReply();
            reply.success();
            Map<String, Object> vars = new LinkedHashMap<>();
            for (int i = 0, size = responses.length; i < size; i++) {
                ThingMessage msg = responses[i];
                Map<String, Object> var = convertToMap(msg);
                vars.putAll(var);
                vars.put("action" + i, var);
            }

            //合并头信息到响应中
            ThingsUtils.mergeHeader(request, reply);


            //使用RPC调用结果作为消息转换上下文，转换模板消息为新的消息
            return responseConverter
                .convert(vars, reply)
                .flux();
        }

        return Flux.empty();
    }

    /**
     * 创建引用消息，用于将上一个节点的返回结果设置到上下文变量中,以便后续操作可以通过此变量获取结果
     *
     * @param request RPC 请求消息
     * @param index   串行执行索引
     * @param all     串行执行结果集
     */
    private ThingMessage createInvokeMessage(ThingMessage request, int index, ThingMessage[] all) {
        if (index > 0 && index <= all.length) {
            ThingMessage pre = all[index - 1];
            if (pre != null) {
                Map<String, Object> vars = convertToMap(pre);
                //将上一个节点的返回结果设置到上下文变量中,以便后续操作可以通过此变量获取结果
                Map<String, Object> context = request.getOrAddHeaderDefault(ThingConstants.contextVar);
                context.putAll(vars);
                context.put("action" + (index - 1), vars);

                //合并头信息到请求中
                ThingsUtils.mergeHeader(pre, request);

            }
        }
        return request;
    }

    /**
     * 将物消息中的数据转换为Map数据
     *
     * @param message 物消息
     */
    protected Map<String, Object> convertToMap(ThingMessage message) {
        return ThingsUtils.messageToContextMap(message);
    }
}
